package com.jugheadzhou.core.zk.helper;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.jugheadzhou.core.zk.ZkClient;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

@Slf4j
public class MythServiceRegistryHelper extends MythServiceBaseZkHelper{
	private static ObjectMapper objectMapper = new ObjectMapper();
    private static SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    /**
     * 当前线程的 InterProcessMutex
     */
    private static final ThreadLocal<InterProcessMutex> INTER_PROCESS_MUTEX = new ThreadLocal<>();
    /**
     * 最大重试次数
     */
    private static final int MAX_RETRY_COUNT = 3;

    private ConnectionStateListener connectionStateListener;
    @Getter
    private final String registryServiceId;
    @Getter
    private final String registryInstanceName;
    private Map<String, String> registryMetaData;
    private static String tempNode = "";
    
    public MythServiceRegistryHelper(ZkClient zkClient, String ip, String port) {
		super(zkClient , ip, port);
		this.registryServiceId = zkClient.getZkProps().getServiceName();
		this.registryInstanceName = getServiceLocalUrl()+"_";
	}

    public String registerInstance(boolean needInstanceNameSequential, Map<String, String> metaData) throws Exception {
        if (metaData == null) {
            metaData = new HashMap<String, String>();
        }
        metaData.put("regTime", sdf.format(new Date()));
        metaData.put("serverIp", getServiceLocalUrl().split(":")[0]);
        this.registryMetaData = metaData;

        String path = this.zkClient.createTempNode(buildInstancePath(registryServiceId, registryInstanceName), objectMapper.writeValueAsString(metaData), needInstanceNameSequential);
        tempNode = path;
        if (this.connectionStateListener == null) {
            this.connectionStateListener = new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {
                    if (connectionState.equals(ConnectionState.RECONNECTED)) {
                        try {
                            if (!tempNode.isEmpty() || !tempNode.equals("")) {
                                boolean exist = zkClient.isExist(tempNode);
                                if (exist) {
                                    log.info("tempNode " + tempNode + " is exist!");
                                } else {
                                    tempNode = zkClient.createTempNode(buildInstancePath(registryServiceId, registryInstanceName),
                                            objectMapper.writeValueAsString(registryMetaData), needInstanceNameSequential);
                                }
                            }
                        } catch (JsonProcessingException e) {
                            log.error("meta info transfer to string is error", e);
                        } catch (Exception e) {
                            log.error("zkClient create temp node is error", e);
                        }
                    }
                }
            };
            zkClient.getClient().getConnectionStateListenable().addListener(connectionStateListener);
        }
        return path;
    }

    public void watchAllServiceInstances(TreeCacheListener listener) throws Exception {
        if (listener == null) {
            throw new Exception("listener can't be null");
        }
        this.zkClient.watchNodeTree(schema, listener);
    }

    public void watchServiceInstances(String serviceId, PathChildrenCacheListener listener) throws Exception {
        if (serviceId == null) {
            throw new Exception("Service id can't be null");
        }
        if (listener == null) {
            throw new Exception("listener can't be null");
        }
        this.zkClient.watchNodeChildren(buildSerivcePath(serviceId), listener);
    }

    private String buildSerivcePath(String serviceId) {
        return schema + ZkClient.SLASH + serviceId;
    }

    private String buildInstancePath(String serviceId, String instanceName) {
        return this.buildSerivcePath(serviceId) + ZkClient.SLASH + instanceName;
    }

    public void lock(String lockKey) {
        try {
            InterProcessMutex interProcessMutex = this.zkClient.createInterProcessMutex(lockKey);
            INTER_PROCESS_MUTEX.set(interProcessMutex);
            interProcessMutex.acquire();
        } catch (Exception e) {
            log.error("lock error lockKey={}", lockKey, e);
            throw new RuntimeException("lock error");
        }
    }

    public boolean lock(String lockKey, long timeout, TimeUnit unit) {
        try {
            InterProcessMutex interProcessMutex = this.zkClient.createInterProcessMutex(lockKey);
            INTER_PROCESS_MUTEX.set(interProcessMutex);
            return interProcessMutex.acquire(timeout, Optional.ofNullable(unit).orElse(TimeUnit.MILLISECONDS));
        } catch (Exception e) {
            log.error("lock error lockKey={}", lockKey, e);
            return false;
        }
    }

    public boolean lockWithRetry(String lockKey, long timeout, TimeUnit unit) {
        try {
            InterProcessMutex interProcessMutex = this.zkClient.createInterProcessMutex(lockKey);
            INTER_PROCESS_MUTEX.set(interProcessMutex);
            for (int i = 0; i < MAX_RETRY_COUNT; i++) {
                if (interProcessMutex.acquire(timeout, Optional.ofNullable(unit).orElse(TimeUnit.MILLISECONDS))) {
                    return true;
                }
                log.info("try lock retryCount={}, lockKey={}", i, lockKey);
            }
            return false;
        } catch (Exception e) {
            log.error("lock error lockKey={}", lockKey, e);
            return false;
        }
    }

    public void unlock() {
        try {
            InterProcessMutex interProcessMutex = INTER_PROCESS_MUTEX.get();
            if (interProcessMutex != null) {
                interProcessMutex.release();
            }
        } catch (Exception e) {
            log.error("unlock error", e);
        } finally {
            INTER_PROCESS_MUTEX.remove();
        }
    }
}
